package com.cw.realtime.dim.function

import com.cw.realtime.common.bean.TableProcessDim
import com.cw.realtime.common.constant.Constant
import com.cw.realtime.common.util.{HbaseUtil, JdbcUtil}
import io.circe.Json
import org.apache.flink.api.common.state.MapStateDescriptor
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction
import org.apache.flink.util.Collector

import scala.collection.mutable

class BcProcessFunction(val bcDesc: MapStateDescriptor[String, TableProcessDim])
  extends BroadcastProcessFunction[Json, TableProcessDim, (Json, TableProcessDim)] {


  private type ResType = (Json, TableProcessDim)
  private type FunctionType = BroadcastProcessFunction[Json, TableProcessDim, ResType]

  val map = mutable.Map.empty[String, TableProcessDim]

  // 提前获取mysql配置表
  override def open(parameters: Configuration): Unit = {
    import JdbcUtil._
    getMysqlConnection.foreach { conn =>
      val lst = queryList[TableProcessDim](conn, "select * from gmall_config.table_process_dim")
//      println(s"queryList: $lst")
      for {
        tbl <- lst
        hbaseConn <- HbaseUtil.getConnection
      }{
        map += tbl.sourceTable -> tbl
        HbaseUtil.createTable(
          hbaseConn, Constant.HBASE_NAMESPACE, tbl.sinkTable, tbl.sinkFamily // 提前创表
        )
        HbaseUtil.closeConnection(Option(hbaseConn))
      }
      closeConnection(conn)
    }
  }

  override def processBroadcastElement(
    value: TableProcessDim, ctx: FunctionType#Context, out: Collector[ResType]): Unit = {

//    println(s"处理广播流:  $value")
    val broadcastState = ctx.getBroadcastState(bcDesc)
    value.op match {
      case "d" =>
        broadcastState.remove(value.sourceTable)
        map -= value.sourceTable
      case _ =>
        broadcastState.put(value.sourceTable, value)
    }
  }

  override def processElement(
    value: Json, ctx: FunctionType#ReadOnlyContext, out: Collector[ResType]): Unit = {
//    println(s"处理左表 ${value}")
    for {
      table <- value.hcursor.downField("table").as[String]
      broadcastState = ctx.getBroadcastState(bcDesc)
      tableProcessDIm = Option(broadcastState.get(table))

    } {
      tableProcessDIm match {
        case Some(t) => out.collect(value -> t)
        case _ => map.get(table).foreach { t => out.collect(value -> t) }
      }
    }
  }


}
